package com.heima.datastructure.blockingqueue;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <p>用两个锁实现阻塞队列</p>
 * <ul>
 *     <li>线程安全
 *         <ul>
 *             <li>两个线程的代码会不会发生交叉</li>
 *             <li>两个线程有没有对同一资源进行增删改操作两个线程的代码会不会发生交叉</li>
 *         </ul>
 *     </li>
 *     <li>线程通信
 *         <ul>
 *             <li>wait()和signal()的等待和通知</li>
 *             <li>由谁来调用这两个方法呢，当然得由锁对象（因为锁会记录此时持有该锁的线程的信息），从而就知道该让哪个线程等待</li>
 *             <li>条件变量Condition继承了锁，而且得根据具体的锁实例创建，它调用await会让持有该锁实例的线程等待并释放该锁</li>
 *             <li>条件变量不仅继承了锁，还扩展了一个存放阻塞线程的队列，一个条件变量对应一个存放同一类被阻塞的线程的队列，所以条件变量要选对</li>
 *             <li>条件变量还必须和可重入锁（父类）一起配套使用</li>
 *             <li>通知过程中，不应该让其他和通知的那个线程所需要的锁一致的线程得到锁</li>
 *         </ul>
 *     </li>
 *     <li>锁的注意问题
 *         <ul>
 *             <li>加锁是实现原子性和同步的手段</li>
 *             <li>使用锁时，尽量不要嵌套地使用锁，容易出现死锁，尽量平级的使用锁</li>
 *             <li>平级的使用锁有的情况还可以优化
 *                 <ul>
 *                      <li>例如本例中，每次offer入队了一个元素之后就要接着唤醒一个poll，</li>
 *                      <li>当offer的次数多于poll的次数或者大部分连续的offer比poll先执行的时候,</li>
 *                      <li>那么offer内部的唤醒阻塞的poll的代码逻辑就是在做无用功，还一直占用poll线程的锁，</li>
 *                      <li>所以思考能不能只在必要的情况下offer才去唤醒被阻塞的poll线程，</li>
 *                      <li>进一步思考能不能通过级联的方式唤醒poll阻塞队列，</li>
 *                      <li>也就是当唤醒的poll进程执行结束后，看队列还有没有元素，如果有，让唤醒的poll线程去唤醒下一个被阻塞的poll线程</li>
 *                      <li>这样一来，效率就又得到提升了，以poll为主体在分析一波</li>
 *                 </ul>
 *             </li>
 *         </ul>
 *     </li>
 * </ul>
 * @author 勾新杰
 * @version 1.0
 * @date 2024/9/4 22:32
 */
public class BlockingQueue2<E> implements BlockingQueue<E>, Iterable<E> {
    private final E[] arr; // 数组
    private int head; // 头指针，指向第一个元素
    private int tail; // 尾指针，指向最后一个元素的下一个元素
    private final AtomicInteger size = new AtomicInteger(); // 当前队列的大小，final修饰的变量是对象的引用，限制变量指向对象的指向不能改变，但是对象内部的值是可以变的，原子性是指代码要么全部执行完，要么不执行，也就是说此线程在修改size时，不会被线程切换打断，不会中断或者回滚，只有修改完后才能切换线程

    // 1、ReentrantLock的可重入是指一个线程可以多次获得该锁，允许在有锁的情况下再次请求锁而不会发生死锁
    // 2、ReentrantLock可以指定公平锁还是非公平锁，公平锁就是按照顺序来获取锁，非公平锁相反，但是后者的性能更好
    // 3、ReentrantLock可以尝试非阻塞的获取锁，即如果获取到的锁不可用就直接返回false，而不是阻塞等待，避免不必要的等待
    // 4、ReentrantLock可中断的锁获取，即线程在等待锁时可以响应中断请求，避免一直被阻塞，一直等不到锁，synchronized就没有这个功能
    // 5、ReentrantLock支持条件变量，提供一个或多个Condition对象，用于线程间的协调，Condition对象提供了非常灵活的条件等待和通知机制
    private final ReentrantLock offerLock = new ReentrantLock(); // 对offer方法提供的单独的一把锁，实现入队方法的原子性，多线程时不会和自己代码发生交叉，如果offer和poll用同一把锁，就会发生这样一个事情：offer拿到锁，poll就会陷入锁阻塞态，只能等其释放锁，反之同理
    private final ReentrantLock pollLock = new ReentrantLock(); // 对poll方法提供的单独的一把锁，实现出队方法的原子性，多线程时不会和自己代码发生交叉，如果offer和poll用同一把锁，就会发生这样一个事情：offer拿到锁，poll就会陷入锁阻塞态，只能等其释放锁，反之同理

    // 1、Condition是与ReentrantLock紧密相关的一个类，用于实现线程间的协调，一个ReentrantLock可以有多个条件变量
    // 2、Condition可以实现指定条件下线程的唤醒和等待
    // 3、根据条件阻塞的线程可以在Condition上等待，直到其他线程调用该Condition的signal方法或者signalAll来通知唤醒它
    // 4、它继承了 ReentrantLock 的所有特性，包括可重入性、公平性、尝试非阻塞地获取锁以及可中断的锁获取等。
    // 5、await和signal必须是在同步代码块内调用
    private final Condition offerWait = offerLock.newCondition(); // 条件变量：每个因为链表为空导致阻塞的poll线程都加入到emptyWait中等待其他线程调用emptyWait的signal方法来唤醒它
    private final Condition pollWait = pollLock.newCondition(); // 条件变量：每个因为链表为满导致阻塞的offer线程都加入到fullWait中等待其他线程调用fullWait的signal方法来唤醒它

    /**
     * 用构造方法实现初始化
     *
     * @param capacity 容量
     */
    @SuppressWarnings("all")
    public BlockingQueue2(int capacity) {
        arr = (E[]) new Object[capacity];
    }

    /**
     * 元素入队
     *
     * @param value 入队元素
     * @throws InterruptedException 中断异常
     */
    @Override
    public void offer(E value) throws InterruptedException {
        offerLock.lockInterruptibly(); // 可响应中断请求的锁，保证不会被一直阻塞
        int prevSize; // offer入队之前的队列的大小
        try {
            while (isFull()) { // 防止虚假唤醒，就是唤醒了不是第一个抢到锁的，可能有线程已经调用了offer方法导致队列为满
                offerWait.await(); // 阻塞当前线程，并释放锁，确保当前线程等待期间不持有锁，当此线程被poll线程唤醒的时候，就会先去抢锁，然后再从这行代码继续往下执行，就进入了while循环
            }
            // offer只操作head指针，poll只操作tail指针，所以head和tail线程安全（对于各自而言锁只有一把，这部分代码不会交叉）
            arr[tail] = value;
            tail = inc(tail);
            // 因为offer和poll是分别用的锁，所以只能保证offer和offer之间和poll和poll之间的线程安全，但是offer和tail都会操作size，size会有线程安全问题（这部分代码会交叉）
            prevSize = size.getAndIncrement(); // 先获取再自增，有原子性
            if (prevSize + 1 < arr.length){ // 当offer之后，队列中还有空余位置可以插入元素，则唤醒一个offer线程
                offerWait.signal();
            }
        } finally {
            offerLock.unlock(); // 解锁，有异常的时候也要解锁
        }
        // 有元素入队后就说明队列中有一个元素了，也就是说生产者线程offer生产了一个元素，那么就该唤醒一个消费者线程poll
        // 平级的使用锁
        if (prevSize == 0) { // offer线程只在必要的情况下才唤醒一个poll线程，其余就让唤醒后的poll自己去唤醒其他阻塞的poll线程，级联，这里的必要情况就是offer之前队列为空（因为poll就是因为队列为空才阻塞的）
            pollLock.lockInterruptibly(); // 一直等不到锁，可以响应中断请求，不等了
            try {
                pollWait.signal(); // 唤醒一个阻塞线程，不释放锁
            } finally {
                pollLock.unlock();
            }
        }
    }

    /**
     * 阻塞队列入队
     *
     * @param value   入队元素
     * @param timeout 当前线程阻塞后等待时间上限，限制单位是毫秒
     * @return 当前线程阻塞等待总时间超过timeout，或者说timout毫秒之后队列还是满，则返回false返回false，其余肯定入队成功，返回true
     * @throws InterruptedException 中断异常
     */
    @Override
    public boolean offer(E value, long timeout) throws InterruptedException {
        offerLock.lockInterruptibly(); // 可响应中断请求的锁，保证不会被一直阻塞
        int prevSize;
        try {
            long remain = toNanos(timeout);
            while (isFull()) { // 防止虚假唤醒，就是唤醒了不是第一个抢到锁的，可能有线程已经调用了offer方法导致队列为满
                if (remain <= 0) { // 已经判断了队列是满并且剩余等待时间已经无了，再等待没意义，返回false
                    return false;
                }
                // 等待时间未被耗尽时，如果被其他线程唤醒，而且没有抢到锁，而且其它生产者线程offer把队列装满了，那么此生产者线程又会被阻塞，所以要保证阻塞的总时间不超过timeout，而不是单次等待
                remain = offerWait.awaitNanos(remain);// 阻塞当前线程，并限制等待时间，返回的是剩余的等待时间，并释放锁，确保当前线程等待期间不持有锁，当此线程被poll线程唤醒的时候，就会先去抢锁，然后再从这行代码继续往下执行，就进入了while循环
            }
            arr[tail] = value;
            tail = inc(tail);
            prevSize = size.getAndIncrement();
            if (prevSize + 1 < arr.length){ // 入队后还有空余位置可以插入
                offerWait.signal();
            }
        } finally {
            offerLock.unlock(); // 解锁，有异常的时候也要解锁
        }
        // 有元素入队后就说明队列中有一个元素了，也就是说生产者线程offer生产了一个元素，那么就该唤醒一个消费者线程poll
        // 平级的使用锁
        if (prevSize == 0) {
            pollLock.lockInterruptibly();
            try {
                pollWait.signal(); // 通知一个正在排队等待的poll线程，通知期间其他poll线程要等待，再公平竞争锁
            } finally {
                pollLock.unlock();
            }
        }
        return true;
    }

    /**
     * 获取队尾元素并删除
     *
     * @return 队尾元素
     * @throws InterruptedException 中断异常
     */
    @Override
    public E poll() throws InterruptedException {
        pollLock.lockInterruptibly();
        int prevSize; // poll之前队列的大小
        E value; // 如果队列一直为空，此poll线程会一直被阻塞，直到队列不为空或者接到一个中断请求进行处理或者抛出异常
        try {
            while (isEmpty()) { // 防止假唤醒
                pollWait.await(); // 阻塞当前线程，并释放该线程持有的锁，将当前线程加入到emptyWait等待，当被生产者线程offer唤醒的时候，应先抢到锁，再从当前代码继续往下执行，进行while循环
            }
            value = arr[head];
            arr[head] = null; // 垃圾回收
            head = inc(head);
            prevSize = size.getAndDecrement(); // 先获取再自减，具有原子性
            if (prevSize - 1 > 0){ // 当当前poll线程出队结束后，队列剩余元素>0则唤醒等待的poll线程
                pollWait.signal();
            }
        } finally {
            pollLock.unlock();
        }
        // 当消费者线程poll消费了一个元素，就应该唤醒一个生产者线程offer
        // 平级的使用锁
        if (prevSize == arr.length) { // 只在必要的情况下才唤醒一个阻塞的offer线程，然后让唤醒的offer线程决定是否要继续唤醒下一个offer线程，级联，必要条件就是poll之前队列是满（因为offer线程就是因为队列为满才阻塞的）
            offerLock.lockInterruptibly();
            try {
                offerWait.signal();
            } finally {
                offerLock.unlock();
            }
        }
        return value;
    }

    /**
     * 出队
     *
     * @param timeout 时间上限
     * @return 被阻塞timeout时间后或者说timeout毫秒时间后如果队列还是空，就返回null，否则返回出队元素
     * @throws InterruptedException 中断异常
     */
    @Override
    public E poll(long timeout) throws InterruptedException {
        pollLock.lockInterruptibly();
        int prevSize;
        E value;
        try {
            long remain = toNanos(timeout);
            while (isEmpty()) {
                if (remain <= 0) {
                    return null;
                }
                remain = pollWait.awaitNanos(remain);
            }
            // offer只操作head指针，poll只操作tail指针，所以head和tail线程安全（对于各自而言锁只有一把，这部分代码不会交叉）
            value = arr[head];
            arr[head] = null;
            head = inc(head);
            // 因为offer和poll是分别用的锁，所以只能保证offer和offer之间和poll和poll之间的线程安全，但是offer和tail都会操作size，size会有线程安全问题（这部分代码会交叉）
            prevSize = size.getAndDecrement();
            if (prevSize - 1 > 0){ // 出队之后还有元素可以出队
                pollWait.signal();
            }
        } finally {
            pollLock.unlock();
        }
        // poll消费者线程删除了队列中的一个元素，就应该唤醒一个生产者线程
        // 平级的使用锁
        if (prevSize == arr.length) { // 只在必要的情况下才去唤醒一个被阻塞的offer线程，然后让唤醒的offer线程元素入队后看是否还有空位置可以插入，如果有，就唤醒一个offer线程，出队之前队列是满的就是必要条件（因为offer就是因为队列为满才阻塞）
            offerLock.lockInterruptibly();
            try {
                offerWait.signal();
            } finally {
                offerLock.unlock();
            }
        }
        return value;
    }

    /**
     * 判断是否为空
     *
     * @return 为空返回true
     */
    public boolean isEmpty() {
        return size.get() == 0;
    }

    /**
     * 判断是否为满
     *
     * @return 为满返回true
     */
    public boolean isFull() {
        return size.get() == arr.length;
    }

    /**
     * 索引自增操作
     *
     * @param index 索引
     * @return 自增后的结果
     */
    private int inc(int index) {
        return ++index == arr.length ? 0 : index;
    }

    /**
     * 索引自减操作
     *
     * @param index 索引
     * @return 自减后的结果
     */
    private int dec(int index) {
        return --index == -1 ? arr.length - 1 : index;
    }

    /**
     * 将毫秒转化为纳秒
     * @param milliSeconds 毫秒
     * @return 纳秒
     */
    private long toNanos(long milliSeconds){
        if (milliSeconds < 0){
            throw new IllegalArgumentException("参数"+milliSeconds+"不能为负");
        }
        return TimeUnit.MILLISECONDS.toNanos(milliSeconds);
    }

    /**
     * 迭代器
     *
     * @return 迭代器
     */
    @Override
    public Iterator<E> iterator() {
        return new Iterator<>() {
            int index = head;
            int cnt = 0;

            @Override
            public boolean hasNext() {
                return cnt < size.get();
            }

            @Override
            public E next() {
                E value = arr[index];
                index = inc(index);
                cnt++;
                return value;
            }
        };
    }

    /**
     * 重写toString方法
     *
     * @return 字符串
     */
    @Override
    public String toString() {
        StringBuilder builder = new StringBuilder();
        builder.append("[");
        int index = head;
        for (int i = 0; i < size.get(); i++) {
            builder.append(arr[index].toString());
            if (index != dec(tail)) {
                builder.append(" ");
            }
            index = inc(index);
        }
        builder.append("]");
        return builder.toString();
    }
}
